package com.beifeng.hadoop.hdfs;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;



public class WedPvMapreduce extends Configured implements Tool{
	
	
	public static class WebPvMap extends Mapper<LongWritable,Text,IntWritable,IntWritable>{

		private IntWritable mapoutputKey = new IntWritable();
		private IntWritable mapoutputValue = new IntWritable(1);
		
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			
			//linue value
			String lineValue = value.toString();
			
			//spilt
			String[] values =  lineValue.split("\t");
			
			//判断得到的values是否大于30  (整个字段是36)
			
			if (30 > values.length) {
				
				context.getCounter("WebPvMapreduce_counts", "limit>30").increment(1L);
				
				return ;
			}
			
			//url
			String urlvalue = values[1];
			
			if (StringUtils.isBlank(urlvalue)) {
				
				context.getCounter("WebPvMapreduce_counts", "url_isblank").increment(1L);
				return ;
			}
			
			String provinceID = values[23];
			
			if (StringUtils.isBlank(provinceID)) {
				context.getCounter("WebPvMapreduce_counts", "provinceid_isblank").increment(1L);
				return ;
			}
			
			Integer proviceInteger  = Integer.MAX_VALUE;
			
			try {
				
				proviceInteger  =  Integer.valueOf(provinceID);
				
			} catch (Exception e) {
				
				context.getCounter("WebPvMapreduce_counts", "provinceid_notnumber").increment(1L);
		
			}
			
			mapoutputKey.set(proviceInteger);
			context.write(mapoutputKey, mapoutputValue);
			
		}		
		
	}
	
	
	public static class WebPvReduce extends Reducer<IntWritable,IntWritable, IntWritable,IntWritable>{

		private IntWritable reduceputValue = new IntWritable();
		
		
		@Override
		protected void reduce(IntWritable key, Iterable<IntWritable> values,Context context)
				throws IOException, InterruptedException {
			
			
			int sum =0;
			
			for (IntWritable value : values) {
				
				sum+=value.get();			
				
			}
			
			reduceputValue.set(sum);
			
			context.write(key, reduceputValue);
		}
		
	}


	public int run(String[] args) throws Exception {
		
		Configuration configuration = this.getConf();
		
		Job job = Job.getInstance(configuration,this.getClass().getSimpleName());
		
		job.setJarByClass(WedPvMapreduce.class);
		
		//inputPath
		
		Path path = new Path(args[0]);
		FileInputFormat.addInputPath(job, path);
		
		//outPath
		
		Path outPath = new Path(args[1]);
		//输出目录如果存在则删除
		FileSystem fsh = outPath.getFileSystem(configuration);
		if (fsh.exists(outPath)) {
			fsh.delete(outPath, true);
		}
		
		FileOutputFormat.setOutputPath(job, outPath);
		
		//mapper
		
		job.setMapperClass(WebPvMap.class);
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		
		//reduce
		
		job.setReducerClass(WebPvReduce.class);
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(IntWritable.class);
		
		//提交job
		
		boolean  isSuccess = job.waitForCompletion(true);
		
		
		return isSuccess ? 0 : 1;	
		
		
	}
	
		public static void main(String[] args) throws Exception {
		
		
		Configuration conf = new Configuration();
		args = new String[]{"hdfs://bigdata01:8020/input/2015082818",
				"hdfs://bigdata01:8020/output_PV"};
		
		int status  =  ToolRunner.run(conf, new WedPvMapreduce(), args);
		System.out.println(status);
	}
	

}
